package com.ruoyi.mqtt.service.impl;

import com.ruoyi.common.exception.base.BaseException;
import com.ruoyi.mqtt.domain.bo.SysMqttBo;
import com.ruoyi.mqtt.gateway.MqttGateway;
import com.ruoyi.mqtt.service.ISysMqttService;
import lombok.RequiredArgsConstructor;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.stereotype.Service;

import java.util.*;

/**
 * mqtt 服务实现类
 *
 * @author 曾午
 * @date 2023/05-11
 */
@Service
@RequiredArgsConstructor
public class ISysMqttServiceImpl implements ISysMqttService {

    private final MqttPahoMessageDrivenChannelAdapter adapter;

    private final MqttGateway mqttGateway;

    /**
     * 获取当前所有topic列表
     */
    @Override
    public List<Map<String, Object>> getTopics() {

        List<Map<String, Object>> result = new ArrayList();

        String[] topics = adapter.getTopic();
        int[] qos = adapter.getQos();

        if (topics != null && topics.length > 0) {
            for (int i = 0; i < topics.length; i++) {
                Map<String, Object> map = new HashMap<>();
                map.put("topic", topics[i]);
                map.put("qos", qos[i]);
                result.add(map);
            }
        }
        return result;
    }


    /**
     * 发送消息到指定topic
     */
    @Override
    public void sendMsgToTopic(SysMqttBo mqttBo) {
        mqttGateway.sendToMqtt(mqttBo.getTopic(), mqttBo.getQos(), mqttBo.getData());
    }

    /**
     * 动态新增topic
     */
    @Override
    public void addTopic(SysMqttBo mqttBo) {
        String[] topics = adapter.getTopic();
        if (!Arrays.asList(topics).contains(mqttBo.getTopic())) {
            adapter.addTopic(mqttBo.getTopic(), mqttBo.getQos());
        } else {
            throw new BaseException("当前主题已订阅！");
        }
    }

    /**
     * 动态移除topic
     */
    @Override
    public void removeTopic(SysMqttBo mqttBo) {
        adapter.removeTopic(mqttBo.getTopic());
    }


}
